package stream;

import javafx.concurrent.Worker;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class wordcount {
    public static void main(String[] args) throws Exception {
        //执行实时计算datastream,创建一个运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //创建输入流
        DataStreamSource<String> source =  env.socketTextStream("192.168.43.51",1234);
    //
        source.flatMap(new FlatMapFunction<String, wordprint>() {
            public void flatMap(String value, Collector<wordprint> out) throws  Exception{
                String[] words = value.split(" ");
                for (String w: words){
                    out.collect(new wordprint(w,1));
                }
                System.out.println("------------");
            }
        }).keyBy("world").sum("count").setParallelism(4)
                .print();
        env.execute("worlf count");


    }
//    创建一个内部类
    public static class wordprint{
        private String world;
        private int count;
        public wordprint(){}

        public wordprint(String world,int count){
            this.world=world;
            this.count=count;
        }

        @Override
        public String toString() {
            return "wordprint{" + "world='" + world + '\'' + ", count=" + count + '}';
        }

        public String getWorld() {
            return world;
        }

        public void setWorld(String world) {
            this.world = world;
        }

        public int getCount() {
            return count;
        }

        public void setCount(int count) {
            this.count = count;
        }
}
}
